package com.sita.redsun.connector.server;

import com.sita.redsun.connector.handler.DecoderHandler;
import com.sita.redsun.connector.handler.EchoServerHandler;
import com.sita.redsun.connector.handler.EncoderHandler;
import com.sita.redsun.connector.handler.MqttHandler;
import com.sita.redsun.connector.mqtthandler.MQTTDecoder;
import com.sita.redsun.connector.mqtthandler.MQTTEncoder;
import com.sita.redsun.util.UnCatchExceptionUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by zhebinwang on 2017/3/28.
 */
public class ServerBase implements  Runnable{

    private ServerBootstrap b;
    private ExecutorService executorService ;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    public static void main(String[] args){
        ServerBase serverBase=new ServerBase();
        serverBase.init();

    }

    public ServerBase() {
    }

    public void init(){

        executorService = Executors.newSingleThreadExecutor();
        // Configure the server.
        UnCatchExceptionUtil.declare();
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        try {
            b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 50)
                    //.handler(new LoggingHandler(LogLevel.ERROR))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new IdleStateHandler(30,30,1800));
                    //ch.pipeline().addLast("decoder", new DecoderHandler());
                    //ch.pipeline().addLast("encoder", new EncoderHandler());
                    ch.pipeline().addLast(new MQTTDecoder());
                    ch.pipeline().addLast(new MQTTEncoder());

                    //ch.pipeline().addLast(  new EchoServerHandler());
                    ch.pipeline().addLast(  new MqttHandler());

                    //ch.pipeline().addLast(new MqttEncoder());
                   // ch.pipeline().addLast(new MqttHandler());
                }
            });

           /* ch.pipeline().addLast("log", new LoggingHandler(LogLevel.DEBUG));
            // 心跳检测
            ch.pipeline().addLast("idle", new IdleStateHandler(config.readTimeout, config.writeTimeout, config.heartbeatTimeout));
            ch.pipeline().addLast("decoder", new PacketDecoder());
            ch.pipeline().addLast("encoder", new PacketEncoder());
            ch.pipeline().addLast("handler", new ServerHandler());*/

            b.option(ChannelOption.SO_BACKLOG, optionSocketBacklog())
                    .option(ChannelOption.ALLOCATOR, optionByteBufAllocator())
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.SO_KEEPALIVE, false);
            b.childOption(ChannelOption.ALLOCATOR, childOptionByteBufAllocator())
                    .childOption(ChannelOption.SO_RCVBUF, 256)
                    .childOption(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.SO_REUSEADDR, true)
                    .childOption(ChannelOption.SO_KEEPALIVE, false)
                    .childOption(ChannelOption.TCP_NODELAY, true);
            //new MqttMessageDecoder();

            // Start the server.



        }
        catch (Exception e){
            e.printStackTrace();
        }
        finally {
            // Shut down all event loops to terminate all threads.
            //bossGroup.shutdownGracefully();
            //workerGroup.shutdownGracefully();
        }
    }


    private int optionSocketBacklog() {
        return 512;
    }

    private ByteBufAllocator optionByteBufAllocator() {
        return PooledByteBufAllocator.DEFAULT;
    }

    private ByteBufAllocator childOptionByteBufAllocator() {
        return PooledByteBufAllocator.DEFAULT;
    }

    private boolean childOptionSocketKeepAlive() {
        return true;
    }

    @Override
    public void run() {

        System.out.println("启动9");
        try{
            ChannelFuture f = b.bind(6200).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        }catch (Exception e){

            e.printStackTrace();
        }
       /* try {
            ChannelFuture future = bootstrap.bind(getPort()).sync();
            future.addListener(startListener);
            channel = future.channel();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            logger.error("SERVER START ERROR", e.getMessage());
        }*/
        finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

    public void doStart(){
        executorService.submit(this);
        //executorService.execute(this);
    }
}
